package org.example.com.atguigu.exercise;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.rdd.RDD;
import org.apache.spark.sql.catalyst.expressions.In;
import org.junit.Test;
import scala.Tuple2;
import scala.Tuple3;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;

public class SparkCoreExercise implements Serializable {
    public static void main(String[] args) {
        // 读取数据
        SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("org.example.com.atguigu.exercise.SparkCoreExercise");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<String> rdd1 = sc.textFile("E:\\IdeaProject\\bigdata0228\\spark0228\\datas\\user_visit_action.txt");
        // 去除 搜索行为 数据
        JavaRDD<String> rdd2 = rdd1.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String v1) throws Exception {
                // 2019-07-27_73_d79508b4-66bf-4410-a5bb-f67a8831610f_50_2019-07-27 19:47:01_null_13_86_null_null_null_null_1
                String[] split = v1.split("_");
                return split[5].equals("null");
            }
        });
        // 列采集
        JavaPairRDD<String, Tuple3<Integer, Integer, Integer>> rdd3 = rdd2.flatMapToPair(new PairFlatMapFunction<String, String, Tuple3<Integer, Integer, Integer>>() {
            @Override
            public Iterator<Tuple2<String, Tuple3<Integer, Integer, Integer>>> call(String line) throws Exception {
                String[] split = line.split("_");
                String click_id = split[6];
                String order_ids = split[8];

                String pay_ids = split[10];
                ArrayList<Tuple2<String, Tuple3<Integer, Integer, Integer>>> res = new ArrayList<>();
                if (!click_id.equals("-1")) {
                    res.add(new Tuple2<>(click_id, new Tuple3<>(1, 0, 0)));
                    return res.iterator();
                } else if (!order_ids.equals("null")) {
                    res.add(new Tuple2<>(order_ids, new Tuple3<>(0, 1, 0)));
                    return res.iterator();
                } else {
                    res.add(new Tuple2<>(pay_ids, new Tuple3<>(0, 0, 1)));
                    return res.iterator();
                }
            }
        });
        //
        JavaPairRDD<String, Tuple3<Integer, Integer, Integer>> rdd4 = rdd3.reduceByKey(new Function2<Tuple3<Integer, Integer, Integer>, Tuple3<Integer, Integer, Integer>, Tuple3<Integer, Integer, Integer>>() {
            @Override
            public Tuple3<Integer, Integer, Integer> call(Tuple3<Integer, Integer, Integer> v1, Tuple3<Integer, Integer, Integer> v2) throws Exception {
                return new Tuple3<>(v1._1() + v2._1(), v1._2() + v2._2(), v1._3() + v2._3());
            }
        });
        System.out.println(rdd4.collect());


    }
    @Test
    public void test(){
        Integer a1 = 2412;
        Integer a2 = 2412;
        System.out.println(a1==a2);
        System.out.println(a1.equals(a2));
    }
}
